/*异步工作线程的设计：
  1、异步工作使用双缓冲区思想
     外界将任务数据，添加到输入缓冲区中
     异步线程对处理缓冲区中的数据进行处理，若处理缓冲区中没有数据了则交换缓冲区
  实现：
  管理的成员：
    1、双缓冲区（生产、消费）
    2、互斥锁/保证线程安全
    3、条件变量-生产&消费（生产缓冲区没有数据，处理完消费缓冲区数据后就休眠）
    4、回调函数（针对缓冲区中数据的处理接口-外界传入一个函数，告诉异步工作器数据该如何处理）
  提供的操作：
    1、停止异步工作器
    2、添加数据到缓冲区
  私有操作：
    创建线程，线程入口函数中，交换缓冲区，对消费缓冲区数据使用回调函数进行处理，处理完后再次交换
*/

/*实现异步工作器*/

#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__
#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>

namespace bitlog
{
    using Functor = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        ASYNC_SAFE,  // 安全状态，表示缓冲区满了则阻塞，避免资源耗尽的风险
        ASUNC_UNSAFE // 不考虑资源耗尽的问题，无限扩容，常用于测试
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor &cb, AsyncType loop_type = AsyncType::ASYNC_SAFE)
            : _stop(false),            // 不停止的
              _looper_type(loop_type), // 默认安全状态
              _thread(std::thread(&AsyncLooper::threadEntry, this)), _callBack(cb)
        {
        }
        ~AsyncLooper()
        {
            stop(); // 停止工作线程
        }
        void stop() // 停止异步工作器的接口
        {
            _stop = true;           // 将退出标志设置为true，停止
            _cond_con.notify_all(); // 唤醒所有的工作线程，是消费者，不断对缓冲区的数据进行处理
            _thread.join();         // 等待工作线程的退出
        }
        void push(const char *data, size_t len)
        {
            // 1、无限扩容--非安全  2、固定大小--生产缓冲区中数据满了就阻塞
            std::unique_lock<std::mutex> lock(_mutex); // 对互斥锁进行管理
            // 条件变量空值，若缓冲区剩余空间大小大于数据长度，则可以添加数据
            if (_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.writeAbleSize() >= len; }); // 安全状态就加上这个代码，不是安全状态，就注释这个代码，真就继续
            // 能够走下来代表满足了条件，可以向缓冲区添加数据
            _pro_buf.push(data, len);
            // 唤醒消费者对缓冲区中的数据进行处理
            _cond_con.notify_one();
        }

    private:
        // 线程的入口函数--对消费缓冲区中的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区
        void threadEntry()
        {

            // while (!_stop)
            while (1)
            {

                // 为互斥锁设置一个生命周期，当缓冲区交换完毕后就解锁（并不对数据的处理过程加锁保护）
                {
                    // 1、判断生产缓冲区有没有数据，有则交换，无则阻塞
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 退出标准被设置，且生产缓冲区已无数据，这时候再退出，否则有可能造成生产缓冲区中有数据，但是没有被完全处理
                    if (_stop && _pro_buf.empty())
                        break;
                    // 若当前是退出前被唤醒，或者有数据被唤醒，则返回真，继续向下运行，否则重新陷入休眠，lock有内部自己的生命周期，走完这个{}后，就会被销毁，就解锁了
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); }); // 为空，则返回真，不为空，则返回假，取反，空为假，非空为真，则返回真，有数据，往走下走,
                                                                            // 被唤醒不单单是有数据，还有当stop为true的时候也往下走
                                                                            // 生产缓冲区中有没有数据，有就向下走

                    _con_buf.swap(_pro_buf); //_con_buf与_pro_buf交换，交换完后，就可以解锁了

                    // 2、唤醒生产者
                    if (_looper_type == AsyncType::ASYNC_SAFE) // 安全状态就唤醒，不是就不唤醒了
                        _cond_pro.notify_all();                // 交换后就唤醒
                }
                // 3、被唤醒后，对消费缓冲区进行数据处理
                _callBack(_con_buf);
                // 4、初始化消费缓冲区
                _con_buf.reset();
            }
        }

    private:
        Functor _callBack; // 具体对缓冲区数据进行处理的回调函数，有异步工作器使用者传入

    private:
        AsyncType _looper_type;            // 要不要进行安全的控制
        std::atomic<bool> _stop;           // 工作器停止标志，可以随时停止异步工作器，是否停止，停止就是ture，
        Buffer _pro_buf;                   // 生产缓冲区
        Buffer _con_buf;                   // 消费缓冲区
        std::mutex _mutex;                 // 互斥锁，保证安全
        std::condition_variable _cond_pro; // 生产等待队列的条件变量
        std::condition_variable _cond_con; // 消费等待队列的条件变量
        std::thread _thread;               // 异步工作器对应的工作线程
    };
}

#endif